package com.dxf.bigdata.D05_spark_again.存储

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * checkpoint 检查点 ,指定落盘路径(persist 虽然也落盘,但是是临时文件会被删除,checkpoint不会)
 *
 *
 */
object checkpoint {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("app")


    sparkConf.set("spark.port.maxRetries","100")

    val sc = new SparkContext(sparkConf)
    sc.setCheckpointDir("checkpointDir")

    //word count
    val rdd: RDD[String] = sc.makeRDD(List("hello spark", "hello word"))

    val splitRDD: RDD[String] = rdd.flatMap(line => {
      println("split ---")
      line.split(" ")
    })

    val mapRDD: RDD[(String, Int)] = splitRDD.map(x => {
      println("map ---")
      (x, 1)
    })

    // 提高效率,缓存mapRDD,否则checkpoint落盘的时候,还会重头执行一遍
    mapRDD.cache()

    //持久化
    mapRDD.checkpoint()

    val value: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)

    // checkpoint落盘会导致再计算一遍?  会的,所以要和cache一起使用
    value.collect().foreach(println)

    println("other ===================")

    // mapRDD 再次使用,RDD是不存储数据的,所以重新计算一遍; mapRDD通过血缘关系重复计算
    val value1: RDD[(String, Iterable[Int])] = mapRDD.groupByKey()

    value1.collect().foreach(println)
    sc.stop()

  }

}
